package com.gudy.engine.bean;

import com.alipay.sofa.jraft.rhea.client.RheaKVStore;
import com.alipay.sofa.jraft.rhea.storage.KVEntry;
import com.alipay.sofa.jraft.util.Bits;
import com.google.common.collect.Lists;
import com.gudy.engine.core.EngineApi;
import lombok.extern.log4j.Log4j2;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang.ArrayUtils;
import thirdpart.bean.CmdPack;
import thirdpart.codec.IBodyCodec;
import thirdpart.order.OrderCmd;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;

@Log4j2
public class CmdPacketQueue {

    private static CmdPacketQueue instance = new CmdPacketQueue();

    private CmdPacketQueue(){}

    public static CmdPacketQueue getInstance(){
        return instance;
    }

    private final BlockingQueue<CmdPack> recvCache = new LinkedBlockingQueue<>();

    public void cache(CmdPack pack){
        recvCache.offer(pack);
    }

    private RheaKVStore orderKVStore;
    private IBodyCodec bodyCodec;
    private EngineApi engineApi;

    public void init(RheaKVStore orderKVStore, IBodyCodec bodyCodec, EngineApi engineApi){
        this.orderKVStore=orderKVStore;
        this.bodyCodec = bodyCodec;
        this.engineApi = engineApi;

        CompletableFuture.runAsync(()->{
            while(true){
                try{
                    CmdPack cmdPack = recvCache.poll(10, TimeUnit.SECONDS);
                    log.info("engine recv :"+cmdPack);
                    if(cmdPack!=null){
                        handle(cmdPack);
                    }
                }catch (Exception e){
                    log.error("msg packet recvcache error ,continue",e);
                }
            }
        });
    }

    private long lastPackNo=-1;
    private void handle(CmdPack cmdPack) throws Exception {
        log.info("recv:{}",cmdPack);
        // nack校验 序号校验
        long packNo = cmdPack.getPackNo();
        if(packNo== lastPackNo+1){
            if(CollectionUtils.isEmpty(cmdPack.getOrderCmds())){
                return;
            }
            for (OrderCmd orderCmd : cmdPack.getOrderCmds()) {
                engineApi.submitCommand(orderCmd);
            }

        }else if(packNo <= lastPackNo){
            //接收到了重复的包
            log.warn("recv duplicate packId :{}",packNo);
        }else{
            // 跳号情况
            log.info("packNo lost from {} to {},begin query from sequencer",lastPackNo+1,packNo);
            // 请求数据
            byte[] firstKey = new byte[8];
            Bits.putLong(firstKey,0,lastPackNo+1);

            byte[] lastKey = new byte[8];
            Bits.putLong(lastKey,0,packNo+1);

            final List<KVEntry> kvEntries = orderKVStore.bScan(firstKey, lastKey);
            if(CollectionUtils.isNotEmpty(kvEntries)){
                List<CmdPack> collect = Lists.newArrayList();
                for (KVEntry entry : kvEntries) {
                    byte[] value = entry.getValue();
                    if(ArrayUtils.isNotEmpty(value)){
                        collect.add(bodyCodec.deserialize(value,CmdPack.class));
                    }
                }
                collect.sort((o1,o2)->(int)(o1.getPackNo()-o2.getPackNo()));
                for (CmdPack pack : collect) {
                    if(CollectionUtils.isEmpty(pack.getOrderCmds())){
                        continue;
                    }
                    for (OrderCmd orderCmd : pack.getOrderCmds()) {
                        engineApi.submitCommand(orderCmd);
                    }
                }
            }else{
                //排队机出错 导致出现了跳号
                lastPackNo=packNo;
            }
        }
    }
}
